package org.idea.qiyu.framework.canal.rocketmq.handler;

import com.alibaba.otter.canal.protocol.CanalEntry;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.idea.qiyu.framework.canal.annotation.CanalListener;
import org.idea.qiyu.framework.canal.handle.AbstractCanalHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @Author linhao
 * @Date created in 7:52 下午 2021/6/21
 */
@CanalListener(table = "t_user", destination = "example")
@Service
public class UserCanalHandler extends AbstractCanalHandler {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    private static String TOPIC = "qiyu-topic";
    private static String TAGS = "qiyu-tag";

    /**
     * 将数据同步到各个子表当中
     *
     * @param rowChange
     */
    @Override
    public void onUpdate(CanalEntry.RowChange rowChange) {
        super.onUpdate(rowChange);
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            System.out.println(rowData);
            byte[] bytes = rowData.toByteArray();
            try {
                Message mqMessage = new Message(TOPIC,TAGS,bytes);
                SendResult sendResult = defaultMQProducer.send(mqMessage);
                System.out.println(sendResult.getSendStatus());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void onDelete(CanalEntry.RowChange rowChange) {
        super.onDelete(rowChange);
        System.out.println(rowChange);
    }

    @Override
    public void onInsert(CanalEntry.RowChange rowChange) {
        super.onInsert(rowChange);
        System.out.println(rowChange);
    }
}
